package org.ztest.rx;

import java.util.Random;
import java.util.concurrent.TimeUnit;

import rx.Observable;
import rx.Observer;
import rx.Subscriber;
import rx.schedulers.Schedulers;

// 响应式编程，观察者模式，由被观察者执行任务，观察者被动接收通知(或者叫执行回调)
public class RXTest
{
    public void test() {
        test4();
        sleep(5);
    }

    public void test2() {
        Observable<Integer> ob = Observable.from(new Integer[] { 1, 2, 3 });
        ob.subscribe((x) -> {
            System.out.println("Action1 " + Thread.currentThread() + " " + x);
        });
        ob.subscribe(new Observer<Integer>() {
            @Override
            public void onCompleted() {
                System.out.println("Observer onCompleted " + Thread.currentThread());
            }

            @Override
            public void onError(Throwable e) {
                System.out.println("Observer onError " + Thread.currentThread());
            }

            @Override
            public void onNext(Integer t) {
                System.out.println("Observer onNext " + Thread.currentThread() + " " + t);
            }
        });
    }

    public void test3() {
        Observable<String> getHtmlPage = getHtmlPage();
        getHtmlPage = getHtmlPage.subscribeOn(Schedulers.io());
        getHtmlPage.subscribe(new Subscriber<String>() {
            @Override
            public void onCompleted() {
                System.out.println("Subscriber onCompleted " + Thread.currentThread());
            }

            @Override
            public void onError(Throwable e) {
                System.out.println("Subscriber onError " + Thread.currentThread());
            }

            @Override
            public void onNext(String t) {
                System.out.println("Subscriber onNext " + Thread.currentThread() + " " + t);
            }
        });
        getHtmlPage.subscribe(new Subscriber<String>() {
            @Override
            public void onCompleted() {
                System.out.println("Subscriber 2 onCompleted " + Thread.currentThread());
            }

            @Override
            public void onError(Throwable e) {
                System.out.println("Subscriber 2 onError " + Thread.currentThread());
            }

            @Override
            public void onNext(String t) {
                System.out.println("Subscriber 2 onNext " + Thread.currentThread() + " " + t);
            }
        });
    }

    public Observable<String> getHtmlPage() {
        return Observable.create(new Observable.OnSubscribe<String>() {
            @Override
            public void call(Subscriber<? super String> t) {
                try {
                    String htmlPage = getStringDelayed();
                    t.onNext(htmlPage);
                    t.onCompleted();
                } catch (Exception e) {
                    e.printStackTrace();
                    t.onError(e);
                }
            }
        });
    }

    public String getStringDelayed() {
        System.out.println("getStringDelayed " + Thread.currentThread());
        sleep(1);
        return "<!DOCTYPE html><!--STATUS OK--> " + new Random().nextInt();
    }

    public void sleep(int seconds) {
        try {
            TimeUnit.SECONDS.sleep(seconds);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    public void test4() {
        Observable.range(1, 10).buffer(3, 3).subscribe(x -> System.out.println(x));
    }
}
